#include <sys/socket.h>
#include <sys/epoll.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <fcntl.h>
#include <unistd.h>
#include <stdio.h>
#include <pthread.h>
#include <errno.h>
#include <signal.h>

#include "conn.h"
#include "interface.h"
#include "debug.h"
#include "conf.h"
#include "timer.h"

ctx_t  g_ctx;


extern int set_reuseaddr(int sockfd);
extern int start_server(int server_id);
extern void* send_fun(void* arg);
extern conn_t* active_connect(int id);
extern conn_t* conn_tab_find_from_node(node_t* node);


int get_local_id(){
	return g_ctx.server_id;
}

 void set_send_callback(SCB scb_fun){
	g_ctx.send_callback_fun = scb_fun;
}

static rval_t send_msg(msg_t* msg){
		conn_t* conn; 
		task_t* task;
		node_t* node;
		rval_t  rval;
		int id;		

		rval.err = 0;
		
		id = msg->hr.to;
		node = find_node_from_id(id);
		if(!node){
			rval.err = EINVAL;
			return rval;
		}
		
		conn = conn_tab_find_from_node(node);
		if(conn==NULL){
			conn = active_connect(id);
		}

		enque(&conn->q_send, &msg->link);
	
		if(get_conn_state(conn) == CONNECTED){
			if(conn->snd_state == IDEL){
				task = new_task(send_fun, (void*)conn); 
				schedule(&g_ctx.tp, task);
				debug_info("new send data task:%llu, conn[%d], conn sndstate[%d]\n",
								 task->id, conn->id, conn->snd_state);
			}
		}

		rval.p = msg;
		return rval;
	
}

static rval_t send_data(int id , char* buf, int len){
		msg_t* msg;
		rval_t  rval;
		rval.err = 0;
		//struct epoll_event ev;
	
		if(!buf ||len ==0){
			rval.err = EINVAL;
			return  rval;
		}
	
		msg = new_msg(REQ_MSG, id, buf, len);

		rval = send_msg(msg);
		if(failed(rval)){
			my_free(msg);
		}
		
		return rval;
}

int send_data_sync(int id , char* buf, int len){
	msg_t* msg;
	rval_t rval;
	
	rval  = send_data(id, buf, len);
	if(failed(rval)){
		return rval.err;
	}
	msg = (msg_t*) rval.p;
	sem_wait(&msg->sem_send);
	my_free(msg);
	return 0;

}

rval_t send_data_async(int id , char* buf, int len){
	return  send_data(id, buf, len);
}

int poll(rval_t rval){
	msg_t* msg;
	msg = (msg_t*) rval.p;
	sem_wait(&msg->sem_send);
	free_msg(msg);
	return 0;
}

 int send_msg_ack(msg_t* msg){
	rval_t rval;
	int len = 1;
	char* buf = my_malloc(len);
	buf[0]='A';
	
	msg_t* ack  = new_msg(ACK_MSG, msg->hr.from, buf, len);
	ack->hr.id = msg->hr.id;
	rval = send_msg(ack);
	if(failed(rval)){
		debug_err("send ack failed! rval err[%d]\n", rval.err);
	}

	return rval.err;
	
}


/*
 * We should free msg space.
*/
 msg_t* recv_data(int block){
	struct list_head* list;
	msg_t* msg = NULL;
	if(block)
		list = deque(&g_ctx.recv_msg_q);
	else
		list = deque_noblock(&g_ctx.recv_msg_q);
	if(list)
		msg = list_entry(list, msg_t, link);
	return msg;
}


int debug_level = WARN_LEVEL;


extern void read_conf(char* file_name);

extern void start_reconn_thread();
extern void stop_reconn_thread();


void singalHandler(int seg){
	debug_info("recevied signal!  [%d]\n", seg);
}

void singalIntHandler(int seg){
	debug_info("recevied signal!  [%d]\n", seg);
	exit(0);
}

void signal_mask(){
	
  	signal(SIGINT, singalIntHandler);
  // 	signal(SIGTERM, singalHandler);
	signal(SIGPIPE, singalHandler);
}

int sys_init(int server_id, SCB send_callback_fun){
	int node_n = 0;

	set_debug_level(INFO_LEVEL);
	printf("debug level:%d\n", get_debug_level());

	log_init("./log");
	debug_info("debug level:%d\n", get_debug_level());

	read_conf(NULL);
    signal_mask();

	g_ctx.send_callback_fun = send_callback_fun;
	
	queue_init(&g_ctx.recv_msg_q,0);
	queue_init(&g_ctx.reconn_q,0);

	init_timer(10240);
	start_server(server_id);
	
	start_epoll_thread();
	start_reconn_thread();
	
	conn_tab_init(&g_ctx.conn_tab);
	
	tpool_init(&g_ctx.tp,"send and receive threads pool", 4);
	
	start(&g_ctx.tp);	
}

int sys_exit(){
	stop_epoll_thread();
	stop_reconn_thread();
	wait_all(&g_ctx.tp);
	destroy_timer();
}
